#include "local_registry_common.hpp"

hloop_t *conn_loop = nullptr;
unpack_setting_t g_unpack_setting = {};
std::vector<st_local_client_info> g_clients_info = {};
std::vector<st_local_service_info> g_services_info = {};
sem_t get_clients_sem;
sem_t get_services_sem;

static void on_close(hio_t *io)
{
    LOG_PRINT_DEBUG("on_close fd[%d] error[%d]", hio_fd(io), hio_error(io));
}

static void on_recv(hio_t *io, void *buf, int readbytes)
{
    LOG_PRINT_INFO("on_recv fd[%d], readbytes[%d]", hio_fd(io), readbytes);
    if (readbytes < (int)LOCAL_REGISTRY_HV_MSG_HEADER_SIZE)
    {
        LOG_PRINT_ERROR("not a complete msg");
        return;
    }

    bool status = false;
    uint32_t real_readbytes = (uint32_t)((uint32_t)readbytes - LOCAL_REGISTRY_HV_MSG_HEADER_SIZE);
    st_local_msg_header *recv_msg_header = (st_local_msg_header *)buf;
    if (real_readbytes != recv_msg_header->msg_len)
    {
        LOG_PRINT_ERROR("msg_len[%u] != real_readbytes[%u]", recv_msg_header->msg_len, real_readbytes);
        return;
    }

    uint8_t *buffer = (uint8_t *)buf;
    pb_istream_t stream = pb_istream_from_buffer(&(buffer[LOCAL_REGISTRY_HV_MSG_HEADER_SIZE]), recv_msg_header->msg_len);
    switch (recv_msg_header->msg_id)
    {
        case local_registry_msg_ctrl_get_clients:
            {
                st_ctrl_get_clients get_clients_info = st_ctrl_get_clients_init_zero;
                status = pb_decode(&stream, st_ctrl_get_clients_fields, &get_clients_info);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }

                if (0 == get_clients_info.more_flag)
                {
                    if (g_clients_info.size() == 0)
                    {
                        LOG_PRINT_INFO("no clients!");
                    }
                    else
                    {
                        for (size_t i = 0; i < g_clients_info.size(); i++)
                        {
                            LOG_PRINT_INFO("client_id[%u], client_pid[%u], client_name[%s], client_status[%u] produce_services_count[%u]",
                                           g_clients_info[i].client_id,
                                           g_clients_info[i].client_pid,
                                           g_clients_info[i].client_name,
                                           g_clients_info[i].client_status,
                                           g_clients_info[i].produce_services_count);
                            for (size_t j = 0; j < g_clients_info[i].produce_services_count; j++)
                            {
                                LOG_PRINT_INFO("service_id[%u], service_type[%u], service_status[%u]",
                                               g_clients_info[i].produce_services[j].service_id,
                                               g_clients_info[i].produce_services[j].service_type,
                                               g_clients_info[i].produce_services[j].service_status);
                            }
                        }
                    }

                    sem_post(&get_clients_sem);
                }
#if 0
            // each recv will print
            LOG_PRINT_INFO("get clients info, client_count[%u], more_flag[%u]", get_clients_info.clients_count, get_clients_info.more_flag);
            for (size_t i = 0; i < get_clients_info.clients_count; i++)
            {
                LOG_PRINT_INFO("client_id[%u], client_pid[%u], client_name[%s], client_status[%u] produce_services_count[%u]",
                               get_clients_info.clients[i].client_id,
                               get_clients_info.clients[i].client_pid,
                               get_clients_info.clients[i].client_name,
                               get_clients_info.clients[i].client_status,
                               get_clients_info.clients[i].produce_services_count);
                for (size_t j = 0; j < get_clients_info.clients[i].produce_services_count; j++)
                {
                    LOG_PRINT_INFO("service_id[%u], service_type[%u], service_status[%u]",
                                   get_clients_info.clients[i].produce_services[j].service_id,
                                   get_clients_info.clients[i].produce_services[j].service_type,
                                   get_clients_info.clients[i].produce_services[j].service_status);
                }
            }
#endif
                for (size_t l = 0; l < get_clients_info.clients_count; l++)
                {
                    g_clients_info.push_back(get_clients_info.clients[l]);
                }
            }
            break;
        case local_registry_msg_ctrl_get_services:
            {
                st_ctrl_get_services get_services_info = st_ctrl_get_services_init_zero;
                status = pb_decode(&stream, st_ctrl_get_services_fields, &get_services_info);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }

                if (get_services_info.more_flag == 0)
                {
                    if (g_services_info.size() == 0)
                    {
                        LOG_PRINT_INFO("no services!");
                    }
                    else
                    {
                        for (size_t i = 0; i < g_services_info.size(); i++)
                        {
                            if (g_services_info[i].has_service)
                            {
                                LOG_PRINT_INFO("service_id[%u], service_type[%u], service_status[%u], listener_clients_count[%u]",
                                               g_services_info[i].service.service_id,
                                               g_services_info[i].service.service_type,
                                               g_services_info[i].service.service_status,
                                               g_services_info[i].listener_clients_count);
                                if (g_services_info[i].has_provide_client)
                                {
                                    LOG_PRINT_INFO("provide_client_id[%u], provide_client_pid[%u], provide_client_name[%s], provide_client_status[%u]",
                                                   g_services_info[i].provide_client.client_id,
                                                   g_services_info[i].provide_client.client_pid,
                                                   g_services_info[i].provide_client.client_name,
                                                   g_services_info[i].provide_client.client_status);
                                }
                                else
                                {
                                    LOG_PRINT_INFO("no provide_client");
                                }

                                if (g_services_info[i].listener_clients_count > 0)
                                {
                                    for (size_t j = 0; j < g_services_info[i].listener_clients_count; j++)
                                    {
                                        LOG_PRINT_INFO("listener_client_id[%u], listener_client_pid[%u], listener_client_name[%s], listener_client_status[%u]",
                                                       g_services_info[i].listener_clients[j].client_id,
                                                       g_services_info[i].listener_clients[j].client_pid,
                                                       g_services_info[i].listener_clients[j].client_name,
                                                       g_services_info[i].listener_clients[j].client_status);
                                    }
                                }
                                else
                                {
                                    LOG_PRINT_INFO("no listener_clients");
                                }
                            }
                        }
                    }

                    sem_post(&get_services_sem);
                }

#if 0
                // each recv will print
                LOG_PRINT_INFO("get services info, services_count[%u], more_flag[%u]", get_services_info.services_count, get_services_info.more_flag);
                for (size_t k = 0; k < get_services_info.services_count; k++)
                {
                    if (get_services_info.services[k].has_service)
                    {
                        LOG_PRINT_INFO("service_id[%u], service_type[%u], service_status[%u], listener_clients_count[%u]",
                                       get_services_info.services[k].service.service_id,
                                       get_services_info.services[k].service.service_type,
                                       get_services_info.services[k].service.service_status,
                                       get_services_info.services[k].listener_clients_count);
                        if (get_services_info.services[k].has_provide_client)
                        {
                            LOG_PRINT_INFO("provide_client_id[%u], provide_client_pid[%u], provide_client_name[%s], provide_client_status[%u]",
                                           get_services_info.services[k].provide_client.client_id,
                                           get_services_info.services[k].provide_client.client_pid,
                                           get_services_info.services[k].provide_client.client_name,
                                           get_services_info.services[k].provide_client.client_status);
                        }
                        else
                        {
                            LOG_PRINT_INFO("no provide_client");
                        }
                        if (get_services_info.services[k].listener_clients_count > 0)
                        {
                            for (size_t j = 0; j < get_services_info.services[k].listener_clients_count; j++)
                            {
                                LOG_PRINT_INFO("listener_client_id[%u], listener_client_pid[%u], listener_client_name[%s], listener_client_status[%u]",
                                               get_services_info.services[k].listener_clients[j].client_id,
                                               get_services_info.services[k].listener_clients[j].client_pid,
                                               get_services_info.services[k].listener_clients[j].client_name,
                                               get_services_info.services[k].listener_clients[j].client_status);
                            }
                        }
                    }
                    else
                    {
                        LOG_PRINT_INFO("no service");
                    }
                }
#endif

                for (size_t l = 0; l < get_services_info.services_count; l++)
                {
                    g_services_info.push_back(get_services_info.services[l]);
                }
            }
            break;
        default:
            {
                LOG_PRINT_ERROR("unknown msg_id[%d]", recv_msg_header->msg_id);
                break;
            }
    }
}

static void on_write(hio_t *io, const void *buf, int writebytes)
{
    (void)buf;
    if (!hio_write_is_complete(io))
    {
        return;
    }

    LOG_PRINT_DEBUG("on_write fd[%d], writebytes[%d]", hio_fd(io), writebytes);
    hio_setcb_read(io, on_recv);
    hio_set_unpack(io, &g_unpack_setting);
    hio_read(io);
}

static void on_connect(hio_t *io)
{
    LOG_PRINT_DEBUG("on_connect fd[%d]", hio_fd(io));
    if (hio_is_connected(io))
    {
        hio_setcb_write(io, on_write);

        // get clients
        uint8_t buffer[LOCAL_REGISTRY_HV_MSG_HEADER_SIZE] = {};
        uint32_t msg_id = local_registry_msg_ctrl_get_clients;
        uint32_t msg_len = 0;
        memcpy(buffer, &msg_id, sizeof(msg_id));
        memcpy(buffer + 4, &msg_len, sizeof(msg_len));
        hio_write(io, buffer, sizeof(buffer));

        // get services
        memset(buffer, 0, sizeof(buffer));
        msg_id = local_registry_msg_ctrl_get_services;
        memcpy(buffer, &msg_id, sizeof(msg_id));
        memcpy(buffer + 4, &msg_len, sizeof(msg_len));
        hio_write(io, buffer, sizeof(buffer));
    }
    else
    {
        LOG_PRINT_ERROR("peer is closed");
        hio_close(io);
    }
}

static void *ctrl_run_thread_func(void *arg)
{
    (void)arg;
    hloop_run(conn_loop);
    return NULL;
}

int main()
{
    std::string conn_localaddr = LOCAL_REGISTRY_SOCKET_PATH;

    if (access(conn_localaddr.c_str(), F_OK) != 0)
    {
        LOG_PRINT_ERROR("daemon is not running!");
        return -1;
    }

    g_unpack_setting.mode = UNPACK_BY_LENGTH_FIELD;
    g_unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
    g_unpack_setting.body_offset = sizeof(st_local_msg_header);
    g_unpack_setting.length_field_offset = sizeof(uint32_t);
    g_unpack_setting.length_field_bytes = sizeof(uint32_t);
    g_unpack_setting.length_adjustment = 0;
    g_unpack_setting.length_field_coding = ENCODE_BY_LITTEL_ENDIAN;

    sem_init(&get_clients_sem, 0, 0);
    sem_init(&get_services_sem, 0, 0);

    conn_loop = hloop_new(HLOOP_FLAG_AUTO_FREE);
    if (nullptr == conn_loop)
    {
        LOG_PRINT_ERROR("hloop_new fail!");
        return -1;
    }

    hio_t *connio = hio_create_socket(conn_loop, conn_localaddr.c_str(), -1, HIO_TYPE_SOCK_STREAM, HIO_CLIENT_SIDE);
    if (nullptr == connio)
    {
        LOG_PRINT_ERROR("hio_create_socket fail!");
        return -1;
    }

    unlink(LOCAL_REGISTRY_CTRL_SOCKET_PATH);
    struct sockaddr_un ctrl_addr = {};
    ctrl_addr.sun_family = AF_UNIX;
    snprintf(ctrl_addr.sun_path, sizeof(ctrl_addr.sun_path), "%s", LOCAL_REGISTRY_CTRL_SOCKET_PATH);
    if (bind(hio_fd(connio), (struct sockaddr *)&ctrl_addr, (socklen_t)SUN_LEN(&ctrl_addr)))
    {
        LOG_PRINT_ERROR("bind %s fail, errno[%d](%s)!", LOCAL_REGISTRY_CTRL_SOCKET_PATH, errno, strerror(errno));
        return -1;
    }
    hio_setcb_close(connio, on_close);
    hio_setcb_connect(connio, on_connect);
    hio_set_connect_timeout(connio, 1000);
    if (0 != hio_connect(connio))
    {
        LOG_PRINT_ERROR("hio_connect fail, errno[%d]!", hio_error(connio));
        return -1;
    }

    pthread_t ctrl_run_thread = {};
    pthread_create(&ctrl_run_thread, NULL, ctrl_run_thread_func, NULL);

    sem_wait(&get_clients_sem);
    LOG_PRINT_INFO("get clients done.");
    sem_destroy(&get_clients_sem);

    sem_wait(&get_services_sem);
    LOG_PRINT_INFO("get services done.");
    sem_destroy(&get_services_sem);

    hio_close(connio);
    hloop_stop(conn_loop);
    pthread_join(ctrl_run_thread, NULL);

    return EXIT_SUCCESS;
}